package com.fcat.springboot.mqtt.emqx;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class MqttPushClient {
    private static final Logger logger = LoggerFactory.getLogger(MqttPushClient.class);

    @Autowired
    private PushCallback pushCallback;

    public static MqttClient client;

    public MqttClient getClient() {
        return client;
    }


    public boolean connect(String host, String clientID, String username, String password, int timeout, int keepalive){
        MqttClient client;
        try {
            client = new MqttClient(host, clientID+"-"+System.currentTimeMillis(), new MemoryPersistence());
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(false);
            options.setUserName(username);
            options.setPassword(password.toCharArray());
            options.setConnectionTimeout(0);//
            options.setKeepAliveInterval(0);
            options.setAutomaticReconnect(true);
            MqttPushClient.client = client;
            try {
                client.setCallback(pushCallback);
                //client.connect(options);
                IMqttToken iMqttToken = client.connectWithResult(options);
                boolean complete = iMqttToken.isComplete();
                logger.info("MQTT连接"+(complete?"成功":"失败"));
                return complete;
            } catch (Exception e) {
                e.printStackTrace();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        logger.info("MQTT连接失败");
        return false;
    }
/**
     * 发布，默认qos为0，非持久化
     * @param topic
     * @param pushMessage
     */

    public void publish(String topic,String pushMessage)throws Exception{
        publish(0, false, topic, pushMessage);
    }

    public void publish(int qos,String topic,String pushMessage)throws Exception{
        publish(qos, false, topic, pushMessage);
    }
/**
     * 发布
     * @param qos
     * @param retained
     * @param topic
     * @param pushMessage
     */

    public void publish(int qos,boolean retained,String topic,String pushMessage) throws Exception{
        MqttTopic mTopic = client.getTopic(topic);
        if(null == mTopic){
            logger.debug("topic not exist");
        }
        MqttDeliveryToken token;
        token = mTopic.publish(pushMessage.getBytes(), qos, retained);
        if(!retained){
            token.waitForCompletion();
        }
    }
    public void publish(int qos,String topic,byte[] pushMessage)throws Exception{
        MqttTopic mTopic = client.getTopic(topic);
        if(null == mTopic){
            logger.debug("topic not exist");
        }
        MqttDeliveryToken token;
        token = mTopic.publish(pushMessage, qos, false);
        token.waitForCompletion();
    }

/**
     * 订阅某个主题，qos默认为0
     * @param topic
     */

    public void subscribe(String topic) throws MqttException {
        subscribe(topic,0);
    }


/**
     * 订阅某个主题
     * @param topic
     * @param qos
     */

    public void subscribe(String topic,int qos) throws MqttException {
        client.subscribe(topic, qos);
    }
}
